package org.apache.hadoop.hdfs.server.datanode;

import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class ChunkHandler extends Thread {
	
	private BlockingQueue<Chunk> inboundChunks;
	private List<Chunk> chunksBeingProcessed;
	private long expectedDataLength;
	private long totalDataReceived = 0;
	private AtomicBoolean terminateAfterDealingWithRemainingChunks = new AtomicBoolean(false);
	private int maxChunksPerPacket;
	private ByteBuffer dataToWriteToPacket;
	private BlockTransferThrottler throttler;
	private int bytesPerChecksum;
	private int checksumSize;
	private DataOutputStream out;
	private boolean corruptChecksumOK;
	private CountDownLatch allBlockDataWrittenLatch = new CountDownLatch(1);
	private long totalDataWritten = 0; // Including data + checksum
	private CountDownLatch RunMethodExecutingLatch = new CountDownLatch(1);
	private long blockSenderId;
	
	private static boolean CLASS_DEBUG_MODE = PrototypeUtilities.DETAILED_DEBUG_MODE;
	
	public ChunkHandler(long startOffset, long endOffset, DataOutputStream out, int packetSize, int maxChunksPerPacket, int checksumSize, int bytesPerChecksum, BlockTransferThrottler throttler, boolean corruptChecksumOK, long blockSenderID) {

		if (CLASS_DEBUG_MODE) {
			System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + " New chunk handler created. Max chunks per packet: " + maxChunksPerPacket + " Start off: " + startOffset + " EndOff: " + endOffset);
		}
		 
		 inboundChunks = new ArrayBlockingQueue<Chunk>(maxChunksPerPacket * 2);
		 expectedDataLength = endOffset - startOffset + 1;
		 this.maxChunksPerPacket = maxChunksPerPacket;
		 dataToWriteToPacket = ByteBuffer.allocate(packetSize);
		 this.throttler = throttler;
		 this.bytesPerChecksum = bytesPerChecksum;
		 this.checksumSize = checksumSize;
		 this.out = out;
		 this.corruptChecksumOK = corruptChecksumOK;
		 this.blockSenderId = blockSenderID;
		
	}
	
	public void handleChunk(Chunk chunk) {
		
		if (CLASS_DEBUG_MODE) {
			System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + " Handling chunk");
		}

//		try {
//			RunMethodExecutingLatch.await();
//		} catch (InterruptedException e) {
//			// TODO Auto-generated catch block
//			e.printStackTrace();
//		}
		
		if (!terminateAfterDealingWithRemainingChunks.get()) {
			synchronized (inboundChunks) {

				try {
					while (inboundChunks.remainingCapacity() == 0) {
						inboundChunks.notifyAll();
						inboundChunks.wait();
					}
					inboundChunks.add(chunk);
					//System.out.println("Run method not waiting on inbound chunks");
				}
				catch (InterruptedException ex) {
					ex.printStackTrace();
				}

				if (CLASS_DEBUG_MODE) {
					System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + " Chunk added to queue: Chunks on queue: "+ inboundChunks.size());
				}
				
				totalDataReceived = totalDataReceived + chunk.getData().length;
				
				if (CLASS_DEBUG_MODE) {
					System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + " Total received data: "+ totalDataReceived + " Expected data: " + expectedDataLength);
				}

				if (expectedDataLength <= totalDataReceived) {
					if (CLASS_DEBUG_MODE) {
						System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + " Chunk Handler received all data it is expecting. Will drop new data presented to it.");
					}

					terminateAfterDealingWithRemainingChunks.set(true);
				}
				else {
					if (CLASS_DEBUG_MODE) {
						System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + " Chunk Handler is still expecting more data so will not set terminate flag yet");
					}					
				}
				inboundChunks.notifyAll();
				
				if (CLASS_DEBUG_MODE) {
					System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + " Total data received by chunk handler: " + totalDataReceived);
					System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + " Chunks waiting on inbound queue: " + inboundChunks.size() + "(Space left: " + inboundChunks.remainingCapacity() + ")");
				}

			}
		}
		
		else {
			if (CLASS_DEBUG_MODE) {
				System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + " Not doing anything with chunk as ChunkHandler set to terminate after dealing with existing chunks.");
			}			
		}
		
	}
	
	public void run() {

		
		
		
		try {
			if (CLASS_DEBUG_MODE) {
				System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + " Run method starting");
			}
	
			
			long sequenceNumber = 0;
			long dataBytesWritten = 0;
			totalDataWritten = 0;
			boolean allChunksDealtWith = false;
			while (!allChunksDealtWith) {
				
				synchronized (inboundChunks) {
					
					if (CLASS_DEBUG_MODE) {
						System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + " RunMethod has inboundChunks lock. Inbound chunks size: "+ inboundChunks.size() + " MaxChunksPerpack: "+ this.maxChunksPerPacket + " Terminate set: " + this.terminateAfterDealingWithRemainingChunks.get());
					}
	
					
					
					while ((inboundChunks.size() < this.maxChunksPerPacket) && !terminateAfterDealingWithRemainingChunks.get()) {
						
						if (CLASS_DEBUG_MODE) {
							System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + " Insufficient chunks to process at the moment - will wait");
							System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + " Chunks on queue: " + inboundChunks.size() + " Max chunks per pack: " + this.maxChunksPerPacket + " TerminateAfterRemaingin: " + this.terminateAfterDealingWithRemainingChunks);
						}
	
						
						
						try {
							inboundChunks.notifyAll();
							inboundChunks.wait();
						}
						catch (InterruptedException ex) {
							ex.printStackTrace();
						}
					}
					if (terminateAfterDealingWithRemainingChunks.get()) {
						chunksBeingProcessed = new ArrayList<Chunk>(inboundChunks.size());
						
						if (CLASS_DEBUG_MODE) {
							System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + " Draining all chunks to processing queue");
						}
						
						
						inboundChunks.drainTo(chunksBeingProcessed, inboundChunks.size());
					}
					else {
						chunksBeingProcessed = new ArrayList<Chunk>(maxChunksPerPacket);
	
						if (CLASS_DEBUG_MODE) {
							System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + " Draining " + this.maxChunksPerPacket + " to processing queue");
						}
						
						
						inboundChunks.drainTo(chunksBeingProcessed, maxChunksPerPacket);
//						for (int i = 0; i < maxChunksPerPacket; i++) {
//							try {
//								chunksBeingProcessed.add(inboundChunks.take());
//							}
//							catch (InterruptedException ex) {
//						
//						ex.printStackTrace();
//							}
//						}
				
					}
					
					inboundChunks.notifyAll();
				}
				
				// Now process the chunks that have been pulled off the inbound queue. Outside of synchronization
				// on in-bound queue so that in-bound queue can fill up again.
				
				if (CLASS_DEBUG_MODE) {
					System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + " Processing " + chunksBeingProcessed.size() + " chunks in processing queue");
				}
	
				
				int dataLength = chunksBeingProcessed.size() * bytesPerChecksum;
				int packetLength = dataLength + (chunksBeingProcessed.size() * checksumSize) + 4;
				// Where does 4 come from - is it an extra checksum???
	
	
				if (CLASS_DEBUG_MODE) {
					System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + " Writing header");
				}
	
				
				// Write packet header
				try {
					if (CLASS_DEBUG_MODE) {
						System.out.println("Writing packet length: " + packetLength);
					}
					out.writeInt(packetLength);
					out.writeLong(dataBytesWritten);
					out.writeLong(sequenceNumber);
					if (dataBytesWritten + dataLength >= expectedDataLength) {
						out.writeByte((byte) 1);
					}
					else {
						out.writeByte((byte) 0);
					}
					out.writeInt(dataLength);
					
					
					if (PrototypeUtilities.SUMMARY_DEBUG_MODE) {
						System.out.println("Written packet header on behalf of BlockSender " + this.blockSenderId);
					}
					// Write the checksum now
					// Removed this as it should still attempt to write the checksum out, even if the correct one
					// is OK. The original implementation only writes 0s if an exception occurs when reading the real
					// checksum in.
	//				if (corruptChecksumOK) {
	//					byte[] checksumBytes = new byte[chunksBeingProcessed.size() * checksumSize];
	//					out.write(checksumBytes);
	//				}
	//				else {
					for (Chunk chunk : chunksBeingProcessed) {
						out.write(chunk.getChecksum());
						totalDataWritten = totalDataWritten + checksumSize;
					}
	//				}
					
					if (PrototypeUtilities.SUMMARY_DEBUG_MODE) {
						System.out.println("Written checksum on behalf of BlockSender " + this.blockSenderId);
					}
					
					// Write the data
					long dataWritten = 0;
					for (Chunk chunk : chunksBeingProcessed) {
						out.write(chunk.getData());
						dataWritten = dataWritten + chunk.getData().length; // Just total actual data (not checksum)
						totalDataWritten = totalDataWritten + chunk.getData().length; // Represents overall data (inc checksum)
					}
					
					if (PrototypeUtilities.SUMMARY_DEBUG_MODE) {
						System.out.println("Written chunk data on behalf of BlockSender " + this.blockSenderId);
					}
					
					sequenceNumber++;
					dataBytesWritten = dataBytesWritten + dataWritten;
					
					if (PrototypeUtilities.SUMMARY_DEBUG_MODE) {
						System.out.println("Written " + dataBytesWritten + " bytes of data to BlockSender " + this.blockSenderId);
					}
					
					if (CLASS_DEBUG_MODE) {
						System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + " Done writing packet");
					}
					
					if (terminateAfterDealingWithRemainingChunks.get() && (dataBytesWritten == expectedDataLength)) {
						allChunksDealtWith = true;
					}
					
					
				}
				catch (Exception ex) {
					ex.printStackTrace();
				}
				
				if (throttler != null) {
					System.out.println("Throttling");
					throttler.throttle(packetLength);
				}
				
				
				
			}

			allBlockDataWrittenLatch.countDown();
			
			
			if (CLASS_DEBUG_MODE) {
				System.out.println(PrototypeUtilities.getPreambleString(this.getClass()) + " Exiting RUN method");
			}
		}
		catch (Exception ex) {
			ex.printStackTrace();
		}
		
		
	}
	
	// Waits until all data is written
	public long waitUntilAllDataIsWritten() {
		try {
			allBlockDataWrittenLatch.await();
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		return totalDataWritten;
	}
	
	

}